package com.zpa.redisstreamlistener.processor;

import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RStream;
import org.redisson.api.RedissonClient;
import org.redisson.api.StreamMessageId;
import org.redisson.api.stream.StreamAddArgs;

import java.util.Map;

@Slf4j
public class RedisStreamProducer {

    private final RedissonClient redisson;

    public RedisStreamProducer(RedissonClient redisson) {
        this.redisson = redisson;
    }

    public RedisStreamProducer() {
        this.redisson = RedisStreamListenerProcessor.applicationContext.getBean(RedissonClient.class);
    }

    /**
     * 发布单一消息到Stream
     */
    public StreamMessageId publishMessage(String streamName, String key, String msg) {
        RStream<String, String> stream = redisson.getStream(streamName);

        // 新的添加消息方式
        StreamMessageId msgId = null;
        try {
            msgId = stream.add(StreamAddArgs.entry(key, msg));
        } catch (Exception e) {
            log.error("publishMessage 添加消息失败 streamName={} ", streamName, e);
            throw e;
        } finally {
            log.debug("publishMessage 发布消息 streamName={} key={} msg={} msgId={} ", streamName, key, msg, msgId);
        }
        return msgId;
    }

    /**
     * 发布带特定字段的消息
     */
    public StreamMessageId publishMessage(String streamName, Map<String, Object> fields) {
        RStream<String, Object> stream = redisson.getStream(streamName);

        StreamMessageId msgId = null;
        try {
            msgId = stream.add(StreamAddArgs.entries(fields));
        } catch (Exception e) {
            log.error("publishMessage 添加消息失败 streamName={} ", streamName, e);
            throw e;
        } finally {
            log.debug("publishMessage 发布消息 streamName={} fields={} msgId={} ", streamName, fields, msgId);
        }
        return msgId;
    }

    private static final ObjectMapper objectMapper = new ObjectMapper();

    /**
     * 发布带特定字段的消息
     */
    public <T> StreamMessageId publishMessage(String streamName, T fields) {
        RStream<String, Object> stream = redisson.getStream(streamName);

        StreamMessageId msgId = null;
        try {
            Map<String, Object> map = objectMapper.convertValue(fields, Map.class);
            msgId = stream.add(StreamAddArgs.entries(map));
        } catch (Exception e) {
            log.error("publishMessage 添加消息失败 streamName={} ", streamName, e);
            throw e;
        } finally {
            log.debug("publishMessage 发布消息 streamName={} fields={} msgId={} ", streamName, fields, msgId);
        }
        return msgId;
    }
}